package com.lry;

import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;

@Slf4j
public class ConsumeQueue {

    public static final int CQ_STORE_UNIT_SIZE = 20;

    private final DefaultMessageStore defaultMessageStore;

    private final MappedFileQueue mappedFileQueue;

    private final String topic;

    private final int queueId;

    private final ByteBuffer byteBuffer;

    private final String storePath;

    private final int mappedFileSize;

    private long minLogicOffset;


    public ConsumeQueue(
            final String topic,
            final int queueId,
            final String storePath,
            final int mappedFileSize,
            final DefaultMessageStore defaultMessageStore) {
        this.storePath = storePath;
        this.mappedFileSize = mappedFileSize;
        this.defaultMessageStore = defaultMessageStore;

        this.topic = topic;
        this.queueId = queueId;

        String queueDir = this.storePath
                + File.separator + topic
                + File.separator + queueId;

        this.mappedFileQueue = new MappedFileQueue(queueDir, mappedFileSize, new MappedFileQueue.MappedFileCallback() {
            @Override
            public int calWritePosition(MappedFile mappedFile) {
                return ConsumeQueue.this.calWritePosition(mappedFile);
            }
        });
        this.byteBuffer = ByteBuffer.allocate(CQ_STORE_UNIT_SIZE);
    }

    //8(commit log offset) + 4(commit log msg size) + 8 (tag hashcode)
    public int calWritePosition(MappedFile mappedFile){
        MappedByteBuffer mappedByteBuffer = mappedFile.getMappedByteBuffer();
        ByteBuffer slice = mappedByteBuffer.slice();
        slice.position(0);

        int writePosition = 0;

        long commitLogOffset = -1;
        int commitLogMsgSize = -1;

        while(commitLogMsgSize!=0 && slice.hasRemaining()){
             commitLogOffset = slice.getLong();
             commitLogMsgSize = slice.getInt();
             long commitLogTagHashcode = slice.getLong();

            if(commitLogMsgSize!=0){
                writePosition += CQ_STORE_UNIT_SIZE;
            }
        }
        return writePosition;
    }


    public boolean flush(){
        return flush(0);
    }

    public boolean flush(final int flushLeastPages) {
        return this.mappedFileQueue.flush(flushLeastPages);
    }

    public boolean putMessage(DispatchRequest request){

        //commit log offset(8) + msg size(4) + tag hash code(8)
        this.byteBuffer.flip();//切换到写
        this.byteBuffer.limit(CQ_STORE_UNIT_SIZE);
        this.byteBuffer.putLong(request.getCommitLogOffset());
        this.byteBuffer.putInt(request.getMsgSize());
        this.byteBuffer.putLong(request.getTagsCode());

        final long expectLogicOffset = request.getConsumeQueueOffset() * CQ_STORE_UNIT_SIZE;
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);

        if(null!=mappedFile){
            //如果是第一个创建的文件，cq offset ！=0 写指针位置为0 ，则填充空白
            if (mappedFile.isFirstCreateInQueue() &&
                    request.getConsumeQueueOffset() != 0 &&
                    mappedFile.getWrotePosition() == 0) {
                this.minLogicOffset = expectLogicOffset;
                this.mappedFileQueue.setFlushedWhere(expectLogicOffset);
                this.mappedFileQueue.setCommittedWhere(expectLogicOffset);
                this.fillPreBlank(mappedFile, expectLogicOffset);
                log.info("fill pre blank space " + mappedFile.getFileName() + " " + expectLogicOffset + " "
                        + mappedFile.getWrotePosition());
            }

            return mappedFile.append(this.byteBuffer.array());
        }
        return false;
    }

    public boolean load() {
        boolean result = this.mappedFileQueue.load();
        log.info("load consume queue " + this.topic + "-" + this.queueId + " " + (result ? "OK" : "Failed"));
        return result;
    }

    public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
        int mappedFileSize = this.mappedFileSize;
        long offset = startIndex * CQ_STORE_UNIT_SIZE;
        if (offset >= this.getMinLogicOffset()) {
            MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
            if (mappedFile != null) {
                SelectMappedBufferResult result = mappedFile.select((int) (offset % mappedFileSize));
                return result;
            }
        }
        return null;
    }

    private long getMinLogicOffset() {
        return minLogicOffset;
    }

    private void fillPreBlank(final MappedFile mappedFile, final long untilWhere) {
        ByteBuffer byteBuffer = ByteBuffer.allocate(CQ_STORE_UNIT_SIZE);
        byteBuffer.putLong(0L);
        byteBuffer.putInt(Integer.MAX_VALUE);
        byteBuffer.putLong(0L);

        int until = (int) (untilWhere % this.mappedFileQueue.getMappedFileSize());
        for (int i = 0; i < until; i += CQ_STORE_UNIT_SIZE) {
            mappedFile.append(byteBuffer.array());
        }
    }
}
